package com.hzmg.akkacloud.worker.actors;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.cluster.Cluster;
import com.fasterxml.jackson.core.type.TypeReference;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.enums.Protocol;
import com.hzmg.akkacloud.common.exception.AkkaCloudException;
import com.hzmg.akkacloud.common.model.ServiceInstance;
import com.hzmg.akkacloud.common.model.SystemMetrics;
import com.hzmg.akkacloud.common.request.WorkerHeartbeat;
import com.hzmg.akkacloud.common.response.AskResponse;
import com.hzmg.akkacloud.common.serialize.JsonUtils;
import com.hzmg.akkacloud.worker.background.ReactiveDiscoveryClient;
import com.hzmg.akkacloud.worker.common.AkkaCloudWorkerVersion;
import com.hzmg.akkacloud.worker.common.WorkerRuntime;
import com.hzmg.akkacloud.worker.common.utils.AkkaUtils;
import com.hzmg.akkacloud.worker.common.utils.SystemInfoUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.Map;

/**
 * Worker健康度定时上报
 * 同时获取最新service信息
 *
 * @author zbw
 * @since 2022/7/7
 */
@Slf4j

public class HealthReporterActor extends AbstractActor {
    private final WorkerRuntime workerRuntime;
    private final SystemMetrics systemMetrics;
    private final ActorRef reportRouter;
    private boolean isFirst = true;

    public HealthReporterActor(WorkerRuntime workerRuntime) {
        this.workerRuntime = workerRuntime;
        if (workerRuntime.getWorkerConfig().getSystemMetricsCollector() == null) {
            systemMetrics = SystemInfoUtils.getSystemMetrics();
        } else {
            systemMetrics = workerRuntime.getWorkerConfig().getSystemMetricsCollector().collect();
        }
         /*
            这里现在是对整个注册中心集群的其中一个发送心跳，通过轮询的方式。
            只要有一个注册中心可用 心跳就能被集群成功接收。
            这里创建了一个router
         */
        //注册中心数量
        int totalInstances = workerRuntime.getWorkerConfig().getServerAddress().size();
        try {
            reportRouter = AkkaUtils.createrRouter(getContext(), RemoteConstant.SERVER_ACTOR_NAME, Collections.singletonList(RemoteConstant.SERVER_CONSOLE_ROLE), totalInstances, 0);
        } catch (Exception e) {
            throw new AkkaCloudException(e);
        }
    }

    public static Props props(WorkerRuntime workerRuntime) {
        return Props.create(HealthReporterActor.class, () -> new HealthReporterActor(workerRuntime));

    }

    @Override
    public Receive createReceive() {

        return receiveBuilder()
                .match(AskResponse.class, this::onReceiveAskResponse)
                .matchAny(obj -> onReceiveReportWorkerHealth())
                .build();
    }

    private void onReceiveAskResponse(AskResponse askResponse) {

        try {
            if (askResponse.isSuccess()) {
                ReactiveDiscoveryClient.updateStatus(JsonUtils.parseObject(askResponse.getData(), new TypeReference<Map<String, ServiceInstance>>() {
                }));
            } else {
                log.warn("[ReactiveDiscoveryClient] fetch services failed, reason is {}.", askResponse.getMessage());
            }
        } catch (Exception ignore) {
        }
    }

    public void onReceiveReportWorkerHealth() {
        WorkerHeartbeat heartbeat = new WorkerHeartbeat();
        heartbeat.setSystemMetrics(systemMetrics);
        heartbeat.setWorkerAddress(workerRuntime.getWorkerAddress());
        heartbeat.setWebAddress(workerRuntime.getWebAddress());
        heartbeat.setAppName(workerRuntime.getWorkerConfig().getAppName());
        heartbeat.setAppId(workerRuntime.getAppId());
        heartbeat.setHeartbeatTime(System.currentTimeMillis());
        heartbeat.setVersion(AkkaCloudWorkerVersion.getVersion());
        heartbeat.setProtocol(Protocol.AKKA.name());
        heartbeat.setClient(workerRuntime.getWorkerConfig().getAppName());
        heartbeat.setTag(workerRuntime.getWorkerConfig().getTag());
        Cluster cluster = Cluster.get(getContext().getSystem());
        try{
        if (isFirst) {
            cluster.registerOnMemberUp(() -> {
                reportRouter.tell(heartbeat, getSelf());
                isFirst = false;
            });
            //log.info("I send the message");
        } else {
            reportRouter.tell(heartbeat, getSelf());
        }}
        catch (Exception e){
            throw new AkkaCloudException(e);
        }
    }


}
